/**
 * Copyright (c) Huawei Technologies Co., Ltd. 2022. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/**
 * Description: Implement stream cache producer.
 */
#ifndef DATASYSTEM_CLIENT_STREAM_CACHE_PRODUCER_IMPL_H
#define DATASYSTEM_CLIENT_STREAM_CACHE_PRODUCER_IMPL_H

#include <cstdint>
#include <memory>
#include <string>
#include <utility>

#include "datasystem/client/stream_cache/client_base_impl.h"
#include "datasystem/client/stream_cache/producer_consumer_worker_api.h"
#include "datasystem/common/eventloop/timer_queue.h"
#include "datasystem/common/shared_memory/shm_unit.h"
#include "datasystem/common/stream_cache/stream_data_page.h"
#include "datasystem/common/stream_cache/stream_meta_shm.h"
#include "datasystem/stream/element.h"
#include "datasystem/stream/stream_config.h"
#include "datasystem/utils/optional.h"
#include "datasystem/utils/status.h"

namespace datasystem {
namespace client {
namespace stream_cache {
class ProducerImpl : public ClientBaseImpl, public std::enable_shared_from_this<ProducerImpl> {
public:
    /**
     * @brief Construct Producer.
     * @param[in] streamName The name of the stream.
     * @param[in] producerId The uuid of the producer which is generated by Worker.
     * @param[in] delayFlushTime The time used in automatic flush after send and default is 5ms.
     * @param[in] pageSize The size of the page
     * @param[in] workerApi Used to call worker service through rpc.
     * @param[in] mmapManager Used to receive and mmap fd that passed from worker.
     * @param[in] listenWorker Listening to the worker survival status.
     * @param[in] senderProducerNo The producer number generated by worker.
     * @param[in] enableStreamDataVerification Should data verification be on.
     * @param[in] address Local worker address.
     * @param[in] port Local worker port.
     */
    ProducerImpl(std::string streamName, std::string tenantId, std::string producerId, int64_t delayFlushTime,
                 int64_t pageSize, std::shared_ptr<ProducerConsumerWorkerApi> workerApi,
                 std::shared_ptr<StreamClientImpl> client, MmapManager *mmapManager,
                 std::shared_ptr<client::ListenWorker> listenWorker, const ShmView &workArea, uint64_t maxStreamSize,
                 const DataVerificationHeader::SenderProducerNo senderProducerNo,
                 const bool enableStreamDataVerification, const DataVerificationHeader::Address address,
                 const DataVerificationHeader::Port port, StreamMode streamMode, uint64_t streamNo,
                 bool enableSharedPage, uint64_t sharedPageSize, const ShmView &streamMetaView);

    ~ProducerImpl() override;

    /**
     * @brief Produce send one element of the stream each time.
     * @param[in] element The element that to be written.
     * @param[in] timeoutMs The timeout for the call
     * @return K_OK on success; the error code otherwise.
     *         K_RUNTIME_ERROR: depends on the error message.
     *         K_SC_ALREADY_CLOSED: producer is already closed/inactive.
     */
    Status Send(const Element &element, Optional<int64_t> timeoutMs);

    /**
     * @brief If flush elements are not flushed, the local buffer may keep some elements.
     * And the flush operation will ensure that all elements are written to the stream.
     * @return K_OK on success; the error code otherwise.
     *         K_RUNTIME_ERROR: depends on the error message.
     *         K_SC_ALREADY_CLOSED: producer is already closed/inactive.
     */
    Status Flush();

    /**
     * @brief Close the producer, after close it will not allow new elements to be sent.
     * Calling Close() on an already closed consumer will return K_OK.
     * Flush operations will trigger when the local buffer had not flushed elements.
     * @return Status of the call.
     */
    Status Close();

    /**
     * @brief Reset the producer by cleaning up data and metadata.
     * @return K_OK on success; the error code otherwise.
     *         K_RUNTIME_ERROR: Producer is not in reset state.
     */
    Status Reset();

    /**
     * @brief Set producer state to Reset. Cancel flush timer. During reset it will not allow Send, Flush.
     * @return K_OK on success; the error code otherwise.
     *         K_RUNTIME_ERROR: producer is in close state already.
     */
    Status SetStateToReset();

    /**
     * @brief Resume the producer, allowing Send, Flush and Close again.
     * @return K_OK on success; the error code otherwise.
     *         K_RUNTIME_ERROR: producer is not in reset state.
     */
    Status Resume();

    /**
     * @brief Init producer and create create share memory page.
     * @return Status of the call.
     */
    Status Init() override;

    /**
     * @brief Sets the state of the Producer to CLOSE
     */
    void SetInactive() override;

    /**
     * @brief Log helper. Creates the prefix for log messages.
     * @return The generated log prefix for this Producer.
     */
    std::string LogPrefix() const override;

    /**
     * @brief Get the producer id.
     * @return Producer id.
     */
    const std::string &GetProducerId() const
    {
        return producerId_;
    }

    /**
     * @brief Execute a flush.
     */
    void ExecFlush();

private:
    /**
     * @brief Check the state_ of consumer and return status.
     * @return Status of the call.
     */
    Status CheckNormalState() const override;

    /**
     * @brief Send request to create memory page for element.
     * @param[in] timeoutMs Timeout for the call
     * @return Status of the call.
     */
    Status CreateWritePage(Optional<int64_t> timeoutMs);

    /**
     * @brief Update write page address and shm info after successful write page creation.
     * @param[in] writePage Pointer to the created write page
     * @return Status of the call.
     */
    Status CreatePagePostProcessing(const ShmView &lastPageView, std::atomic<int64_t> &fixPageCount);

    /**
     * @brief Produce send one element of the stream each time.
     * @param[in] element The element that to be written (header + data)
     * @param[in] timeoutMs The timeout for the call
     * @return Status of the call.
     */
    Status SendImpl(const HeaderAndData &element, InsertFlags &flags, Optional<int64_t> timeoutMs);

    /**
     * @brief Locate page to insert, likely the last page.
     * @param[in] timeoutMs The timeout for the call.
     * @param[in] requireNewPage If true, require a CreateWritePage to worker if the cursor's page is the current page.
     * @return Status of the call.
     */
    Status FixPage(Optional<int64_t> timeoutMs, const bool requireNewPage);

    Status UnfixPage();
    Status DelayFlush(bool force = false);
    void ExecAndCancelTimer();
    Status HandleNoSpaceFromInsert(int64_t timeoutMs, const Status &rc);

    /**
     * @brief Set the timer to UnfixPage when producer is idle.
     * @return Status of the call.
     */
    Status SetUnfixPageTimer();

    /**
     * @brief Cancel the timer to UnfixPage.
     */
    void UnsetUnfixPageTimer();

    Status InsertBigElement(const HeaderAndData &element, Optional<int64_t> timeoutMs);

    /**
     * @brief Get the last page view from cursor.
     * @param[out] lastPageView The last page view.
     * @param[out] switchToSharedPage Whether swith to shared page.
     * @return Status of this call.
     */
    Status GetLastPageView(ShmView &lastPageView, bool &switchToSharedPage);

    const std::string producerId_;

    int64_t delayFlushTime_;
    int64_t pageSize_;
    std::shared_ptr<StreamDataPage> writePage_{ nullptr };
    std::unique_ptr<TimerQueue::TimerImpl> delayFlushTimer_;
    std::unique_ptr<TimerQueue::TimerImpl> unfixTimer_;
    std::unique_ptr<WaitPost> unfixWaitPost_{ nullptr };
    std::shared_ptr<ShmUnitInfo> pageUnit_;
    ShmView curView_;
    std::mutex flushMutex_;  // Guarantee FIFO, single on-the-fly flush.
    std::atomic<bool> pageDirty_{ false };
    std::atomic<int64_t> fixPageFromRpc_{ 0 };
    std::atomic<int64_t> fixPageFromWorkArea_{ 0 };
    std::atomic<int64_t> fixPageFromNextPage_{ 0 };
    uint64_t maxStreamSize_;
    uint64_t maxElementSize_;
    const DataVerificationHeader::SenderProducerNo senderProducerNo_;
    const bool enableStreamDataVerification_;
    const DataVerificationHeader::Address address_;
    const DataVerificationHeader::Port port_;
    std::atomic<DataVerificationHeader::SeqNo> lastSendElementSeqNo_{ 0 };
    FirstCallTracer sendTracer_;
    StreamMode streamMode_;
    uint64_t streamNo_;
    bool enableSharedPage_;
    uint64_t sharedPageSize_;

    ShmView streamMetaView_;
    std::unique_ptr<StreamMetaShm> streamMetaShm_;

    friend class Producer;
};
}  // namespace stream_cache
}  // namespace client
}  // namespace datasystem
#endif
